package com.yuhang.service;

import com.yuhang.bean.ApacheLogEvent;
import com.yuhang.bean.PageViewCount;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.net.URL;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.Map;
import java.util.regex.Pattern;


public class HotPagesNetworkFlow {

    public static void main(String[] args) throws Exception {

        // 1. 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 2. 读取数据，创建DataStream
//        URL resource = HotPagesNetworkFlow.class.getResource("/apache.log");
//        DataStreamSource<String> inputStream = env.readTextFile(resource.getPath());
        DataStreamSource<String> inputStream = env.readTextFile("D:\\Utils\\IDEA project\\Flink_java\\HotPagesNetworkFlow\\src\\main\\resources\\apache.log");

        // 3. 转换为POJO，分配时间戳和watermark
        DataStream<ApacheLogEvent> dataStream = inputStream.map(data -> {
            String[] fields = data.split(" ");
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("dd/MM/yyyy:HH:mm:ss");
            long timestamp = simpleDateFormat.parse(fields[3]).getTime();
            return new ApacheLogEvent(fields[0],fields[1],timestamp,fields[5],fields[6]);
        })
                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<ApacheLogEvent>(Time.seconds(1)) {

                    @Override
                    public long extractTimestamp(ApacheLogEvent apacheLogEvent) {
                        return apacheLogEvent.getTimestamp();
                    }
                });

        //定义一个测输出流名称
        OutputTag<ApacheLogEvent> outputTag = new OutputTag<ApacheLogEvent>("late"){};

        //不能定义为dataStream类，没有测输出流的打印
        SingleOutputStreamOperator<PageViewCount> aggregate = dataStream
                .filter(data -> {
                    String regexp = "^((?!\\.(css|js|png|ico)$).)*$";
                    return Pattern.matches(regexp,data.getUrl());
                })
                .filter(data -> "GET".equals(data.getMethod()))
                .keyBy(ApacheLogEvent::getUrl)     //写lamba表达式或者方法引用，就不是tuple类型了
                .timeWindow(Time.minutes(10), Time.seconds(5))
                .allowedLateness(Time.minutes(1))
                .sideOutputLateData(outputTag)   //当一条数据都不属于任何窗口的清空下，才会被丢入测输出流中
                .aggregate(new PageCountAgg(), new PageCountResult());

        aggregate.print("agg");
        aggregate.getSideOutput(outputTag).print("late");


        DataStream<String> resultStream = aggregate.keyBy(PageViewCount::getWindowEnd)
                .process(new TopNHotPage(3));


        resultStream.print();

        env.execute("page hot");

    }

    private static class PageCountAgg implements AggregateFunction<ApacheLogEvent,Long,Long>{
        @Override
        public Long createAccumulator() {
            return 0L;
        }

        @Override
        public Long add(ApacheLogEvent apacheLogEvent, Long aLong) {
            return aLong+1;
        }

        @Override
        public Long getResult(Long aLong) {
            return aLong;
        }

        @Override
        public Long merge(Long aLong, Long acc1) {
            return aLong + acc1;
        }
    }

    private static class PageCountResult implements WindowFunction<Long,PageViewCount,String, TimeWindow>{
        @Override
        public void apply(String url, TimeWindow timeWindow, Iterable<Long> iterable, Collector<PageViewCount> collector) throws Exception {
            collector.collect(new PageViewCount(url,timeWindow.getEnd(),iterable.iterator().next()));
        }
    }

    private static class TopNHotPage extends KeyedProcessFunction<Long, PageViewCount, String> {

        private Integer topSize;

        public TopNHotPage(Integer topSize) {
            this.topSize = topSize;
        }

        //用这个有个小问题就是，如果处理迟到数据，如果状态不清空，那么就会多出一个迟到数据的排名（已有排名另开一个，没有新增），但是如果状态清空
        // 那么之前的排名榜单除了迟到数据那个排名，其他排名都不见了，所以我们要采用一种方式，有针对性的更新或者清除，用map+
//        ListState<PageViewCount> pageViewCountState;
        MapState<String,Long> pageViewCountState;

        @Override
        public void open(Configuration parameters) throws Exception {
//            pageViewCountState = getRuntimeContext().getListState(new ListStateDescriptor<PageViewCount>("page-count-list",PageViewCount.class));
            pageViewCountState = getRuntimeContext().getMapState(new MapStateDescriptor<String,Long>("page-count-list",String.class,Long.class));
        }

        @Override
        public void processElement(PageViewCount pageViewCount, Context context, Collector<String> collector) throws Exception {

//            pageViewCountState.add(pageViewCount);

            pageViewCountState.put(pageViewCount.getUrl(),pageViewCount.getCount());

            context.timerService().registerEventTimeTimer(pageViewCount.getWindowEnd()+1);

            //注册一个一分钟的定时器，用来清空状态
            context.timerService().registerEventTimeTimer(pageViewCount.getWindowEnd() + 60 * 1000L);

        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {

            //先判断是否到了窗口关闭清理时间，如果是，直接清空状态返回
            if(timestamp == ctx.getCurrentKey() + 60 * 1000L){
                pageViewCountState.clear();
                return;
            }
            ArrayList<Map.Entry<String, Long>> pageViewCounts = Lists.newArrayList(pageViewCountState.entries().iterator());

//            pageViewCounts.sort(new Comparator<PageViewCount>() {
//                @Override
//                public int compare(PageViewCount o1, PageViewCount o2) {
//
//                    if (o1.getCount() > o2.getCount()){
//                        return -1;
//                    }else if(o1.getCount() < o2.getCount()){
//                        return 1;
//                    }else {
//                        return 0;
//                    }
//                }
//            });
            pageViewCounts.sort(new Comparator<Map.Entry<String, Long>>() {
                @Override
                public int compare(Map.Entry<String, Long> o1, Map.Entry<String, Long> o2) {
                    if(o1.getValue() > o2.getValue()){
                        return -1;
                    }else if(o1.getValue() < o2.getValue()){
                        return 1;
                    }else{
                        return 0;
                    }
                }
            });

            // 将排名信息格式化成String，方便打印输出
            StringBuilder resultBuilder = new StringBuilder();
            resultBuilder.append("===================================\n");
            resultBuilder.append("窗口结束时间：").append( new Timestamp(timestamp - 1)).append("\n");

            // 遍历列表，取top n输出
            for( int i = 0; i < Math.min(topSize, pageViewCounts.size()); i++ ){
//                PageViewCount currentItemViewCount = pageViewCounts.get(i);
                Map.Entry<String, Long> currentItemViewCount = pageViewCounts.get(i);
                resultBuilder.append("NO ").append(i+1).append(":")
                        .append(" 页面URL = ").append(currentItemViewCount.getKey())
                        .append(" 浏览量 = ").append(currentItemViewCount.getValue())
                        .append("\n");
            }
            resultBuilder.append("===============================\n\n");

            // 控制输出频率
            Thread.sleep(1000L);

            out.collect(resultBuilder.toString());
        }
    }

}
